package com.spiro.scences;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.util.Scanner;

/**
 * Created by Administrator on 2016/10/17 0017.
 */
public class LeaderElection {

    public static final Logger LOG = LoggerFactory.getLogger(LeaderElection.class);
    public static final String LEADER_PATH = "/leader";
    public static final String PROCESS_NAME = ManagementFactory.getRuntimeMXBean().getName();

    private ZooKeeper zooKeeper;

    public LeaderElection(String connectString) {
        try {
            zooKeeper = new ZooKeeper(connectString, 10000, new Watcher() {
                public void process(WatchedEvent event) {
                    LOG.info(event.toString());
                    if (event.getState() == Event.KeeperState.Expired) {
                        LOG.info("Session Expired!");
                        System.exit(-1);
                    }
                }
            });
            LOG.info("process-name -> {}", PROCESS_NAME);
        } catch (IOException e) {
            e.printStackTrace();
            System.exit(-1);
        }
    }

    public void leading() {
        zooKeeper.getData(LEADER_PATH, new Watcher() {
            public void process(WatchedEvent event) {
                leading();
            }
        }, new AsyncCallback.DataCallback() {
            public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) {
                KeeperException.Code code = KeeperException.Code.get(rc);

                if (code == KeeperException.Code.OK) {
                    LOG.info("Get leader ok -> {}", new String(data));
                } else if (code == KeeperException.Code.NONODE) {
                    // Node not exists
                    LOG.info("Node not exists, creating");
                    try {
                        zooKeeper.create(LEADER_PATH, PROCESS_NAME.getBytes(),
                                ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
                        LOG.info("Create LEADER_PATH success.");
                        leading();
                    } catch (KeeperException e) {

                        if (e.code() == KeeperException.Code.NODEEXISTS) {
                            leading();
                        } else {
                            LOG.error(e.getMessage(), e);
                            try {
                                Thread.sleep(1000L);
                            } catch (InterruptedException e1) {
                            }
                            leading();
                        }
                    } catch (InterruptedException e) {
                    }
                } else {
                    LOG.error("rc -> {}, path -> {}, stat -> {}", new Object[] {rc, path, stat});
                    try {
                        Thread.sleep(1000L);
                    } catch (InterruptedException e1) {
                    }
                    leading();
                }
            }
        }, null);
    }

    public void blocking() {
        Scanner scanner = new Scanner(System.in);
        String line = null;
        while ((line = scanner.nextLine()) != null) {
            if ("quit".equals(line)) {
                break;
            }
        }
    }

    public static void main(String[] args) {

//        if (args.length == 0) {
//            System.out.println("No args");
//
//            System.exit(-1);
//        }

        String connectString = "192.168.103.108:2181";
        LeaderElection election = new LeaderElection(connectString);
        election.leading();
        election.blocking();
    }
}
